package com.at.bilibili.service.websocket;

import com.alibaba.fastjson.JSONObject;
import com.at.bilibili.domain.Danmu;
import com.at.bilibili.domain.constant.MQConstant;
import com.at.bilibili.service.DanmuService;
import com.at.bilibili.service.util.RocketMQUtil;
import com.at.bilibili.service.util.TokenUtil;
import com.mysql.cj.util.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {

    // 获取当前类的日志记录
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    // 当前长连接的人数 当前连接WebSocket的人数
    // AtomicInteger保证并发环境下操作具体数据的安全性
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    // ConcurrentHashMap保证线程安全
    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    // 保存与客户端关联的Session
    private Session session;

    // 唯一的ID
    private String sessionId;

    // 用户id
    private Long userId;

    // 获取全局的应用上下文 用于解决多例Bean的注入问题
    private static ApplicationContext APPLICATION_CONTEXT;
    public static void setApplicationContext(ApplicationContext applicationContext){
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }


    @OnOpen // 表示当连接成功时 调用该方法
    public void openConnection(Session session){
        this.sessionId = session.getId();
        this.session = session;
        if (WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId, this);
        }
        else{
            WEBSOCKET_MAP.put(sessionId, this);
            // 在线人数加一
            ONLINE_COUNT.getAndIncrement();
        }
        logger.info("用户连接成功 : " + sessionId +", 当前在线人数为 : " + ONLINE_COUNT.get());
        try {
            // 成功
            this.sendMessage("0");
        }catch (Exception e){
            logger.error("连接异常");
        }
    }

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    @OnClose // 表示当连接关闭 调用该方法
    public void closeConnection(){
        if (WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        logger.info("用户退出 : " + sessionId +", 当前在线人数为 : " + ONLINE_COUNT.get());
    }

    @OnMessage // 表示当发送来消息 调用该方法
    public void onMessage(String message, @PathParam("token") String token){
        logger.info("用户信息 : " + sessionId + ", 报文 : " + message);
        try {
            this.userId = TokenUtil.verifyToken(token);
        }catch (Exception e){}
        if (!StringUtils.isNullOrEmpty(message)){
            try {
                // 群发消息
                for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()){
                    // 每一个客户端要进行如下的操作
                    WebSocketService webSocketService = entry.getValue();
                    // 获取生产者
                    DefaultMQProducer danmusProducer = (DefaultMQProducer) APPLICATION_CONTEXT.getBean("danmusProducer");
                    // 将消息转为JSON格式放入到MQ生产者中进行异步发送
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message", message);
                    jsonObject.put("sessionId", webSocketService.getSessionId());
                    Message msg = new Message(MQConstant.TOPIC_DANMUS, jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    RocketMQUtil.asyncSendMsg(danmusProducer,msg);
                }
                if (this.userId != null){
                    // 保存弹幕到数据库
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService) APPLICATION_CONTEXT.getBean("danmuService");
                    danmuService.asyncAddDanmu(danmu);
                    // 保存弹幕到Redis
                    danmuService.addDanmusToRedis(danmu);
                }
            }catch (Exception e){
                    logger.error("弹幕接收出现问题");
                    e.printStackTrace();
            }
        }
    }

    @OnError // 表示当发生错误时 调用该方法
    public void onError(Throwable error){

    }

    //或直接指定时间间隔，例如：5秒
    @Scheduled(fixedRate=5000)
    private void noticeOnlineCount() throws IOException {
        for(Map.Entry<String, WebSocketService> entry : WebSocketService.WEBSOCKET_MAP.entrySet()){
            WebSocketService webSocketService = entry.getValue();
            if(webSocketService.session.isOpen()){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount", ONLINE_COUNT.get());
                jsonObject.put("msg", "当前在线人数为" + ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }

    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }
}
